

#include "BScheduler.h"
#include "BDefs.h"
#include "BUtils.h"
//#include "BMulticaster.h"
using namespace std;
using namespace blueberry;

BScheduler::BScheduler()
{
	i_ReadTurns = 1024;
	i_SockLogFd = -1;
	z_SockLogPath[0] = 0;
	i_TotMicroSec = 0;
	i_LocalSeconds = -1;
	b_ISSocketLogON = false;
	b_ISSocketLogDump = false;
	//i_HBLen = sprintf(z_HeartBeat, "0%c0\n", MSG_DELIM_DS);
	i_MaxTimers = MAX_TIMERS_LIMIT;
	ap_Timers = new BTimer*[i_MaxTimers];
	for (int i = 1; i < MAX_TIMERS_LIMIT; ++i)// 0 for not repeated timers
	{
		ap_Timers[i] = NULL;
		stack_FreeTimerIDs.push(i);
	}

	ui_TotalDeviation = 0;
	ui_CurrentPos = 1;

	i_EpollFd = epoll_create(MAX_EPOL_EVENTS);
	if (i_EpollFd < 0)
	{
		cout << "Main loop epoll failed! Possible cause: No enough memory" << endl;
		exit(1);
	}
	i_CliAddrSize = sizeof(struct sockaddr_in);
	pstEpollRetEvents = new struct epoll_event[MAX_EPOL_EVENTS];

	ap_EItems = new BItem *[MAX_MAX_CLIENT_COUNT];
	for (int i = 0; i < MAX_MAX_CLIENT_COUNT; ++i)
		ap_EItems[i] = NULL;

}

BScheduler::~BScheduler()
{
}

int BScheduler::SetLimits(unsigned int _uiSoftLimit, unsigned int _uiHardLimit)
{
	rlimit limits;

	limits.rlim_cur = _uiSoftLimit;
	limits.rlim_max = _uiHardLimit;

	setrlimit(RLIMIT_NOFILE, &limits);
	return 0;
}

void BScheduler::Run()
{
	p_HBTimer = CreateTimer(MAX_PULSE_TIMEOUT, true, this);

	UInt uiPollInterval = 1;
	GET_CURR_TIME(tvnow,ui_LastTime,ui_MicroSec);

	while (true)
	{
		//ui_TotalDeviation = 0;
		int iLPos = ui_CurrentPos % MAX_ARRAY_SIZE;
		//cout << ui_CurrentPos << " Current pos " << iLPos << endl;

		TIMER_LIST::iterator ite = ast_Timers[iLPos].begin();
		while (ite != ast_Timers[iLPos].end())
		{
			st_TimerBox * pTimeBox = *ite;
			UInt uiTimerId = pTimeBox->ui_TimerID;

			/*remove iterator from list*/
			TIMER_LIST::iterator iteTemp = ite;
			++ite;
			ast_Timers[iLPos].erase(iteTemp); //remove timer from list
			/*****************************/

			if (IsTimerLive(uiTimerId)) //check for user timer deletion
			{
				BTimer * pTimer = pTimeBox->p_Timer;
				UInt uiNestFire = pTimer->GetNextFire();

				if (uiNestFire == 0) //fire now
				{
					BTimerCallback * pTCB = pTimer->GetTimerCB();
					pTCB->OnTimer(pTimer);//firing timer call back

					if (IsTimerLive(uiTimerId))//check for user timer deletion during call back
					{
						if (pTimer->IsRepeated()) //if repeated timer add to correct slot
						{
							pTimer->SetNextFire(pTimer->GetInterval());
							AddToCorrectSlot(pTimeBox);
						}
						else
						{
							// if timer not repeating
							pTimer->ResetTimerId();
							delete pTimeBox;
							stack_FreeTimerIDs.push(uiTimerId);
						}

					}
					else //delete during call back
					{
						delete pTimeBox;
						stack_FreeTimerIDs.push(uiTimerId);
					}

				}
				else // can not fire now
				{
					AddToCorrectSlot(pTimeBox);
				}
			}
			else //timer has been deleted by user
			{
				delete pTimeBox;
				stack_FreeTimerIDs.push(uiTimerId);
			}
		}
		++ui_CurrentPos;
		//cout << "before FireAll " << ui_CurrentPos << endl;

		GET_CURR_TIME(tvnow,ui_CurrTime,ui_MicroSec);
		ui_TotalDeviation += (ui_CurrTime - ui_LastTime) / 2;

		ui_LastTime = ui_CurrTime;

		if (ui_TotalDeviation > 0)
		{
			//cout << " Deviation " << ui_TotalDeviation << endl;
			UInt uiFRet = FireAll();
			//cout << " FireAll " << uiFRet << endl;
			ui_TotalDeviation += uiFRet / 2;
		}

		uiPollInterval = MAX_EPOL_WAIT;
		UInt uiMax = ui_CurrentPos + MAX_EPOL_WAIT;
		for (UInt ui = ui_CurrentPos; ui < uiMax; ++ui)
		{
			int iTemPos = ui % MAX_ARRAY_SIZE;
			if (ast_Timers[iTemPos].size() > 0)
			{
				uiPollInterval = ui - ui_CurrentPos + 1;
				break;
			}
		}

		UInt uiRet;//reading sockets

		uiRet = ReadSockets(uiPollInterval);
		//cout << " ReadSockets " << uiRet << " uiPollInterval"<<uiPollInterval<< endl;
		if (uiPollInterval < uiRet) //has taken more than  allocated time
		{
			ui_TotalDeviation += (uiRet - uiPollInterval) + 1;
			ui_CurrentPos += uiPollInterval - 1;
		}
		else
		{
			if (uiRet > 0)
				ui_CurrentPos += uiRet - 1;
		}

	}
}

UInt BScheduler::FireAll()
{
	//GET_CURR_TIME(tvnow,ui_LastTime,ui_MicroSec);
	UInt uiTemp = 0;

	for (UInt ui = ui_CurrentPos; ui < (ui_CurrentPos + ui_TotalDeviation); ++ui)
	{
		++uiTemp;
		if (uiTemp == MAX_ARRAY_SIZE)
		{
			ui_TotalDeviation = uiTemp - 1;
			break;
		}
		int iLPos = ui % MAX_ARRAY_SIZE;
		TIMER_LIST::iterator ite = ast_Timers[iLPos].begin();
		while (ite != ast_Timers[iLPos].end())
		{
			st_TimerBox * pTimeBox = *ite;
			UInt uiTimerId = pTimeBox->ui_TimerID;
			if (IsTimerLive(uiTimerId)) //if timer not deleted by user
			{
				lst_FireAll.push_back(pTimeBox);
			}
			else
			{
				stack_FreeTimerIDs.push(uiTimerId);
			}

			++ite;
		}
		ast_Timers[iLPos].clear();
	}

	ui_CurrentPos += ui_TotalDeviation;
	ui_TotalDeviation = 0;

	TIMER_LIST::iterator iteAll = lst_FireAll.begin();
	while (iteAll != lst_FireAll.end())
	{
		st_TimerBox * pTimeBox = *iteAll;
		UInt uiTimerId = pTimeBox->ui_TimerID;

		if (IsTimerLive(uiTimerId))//check for user timer deletion during call back
		{
			BTimer * pTimer = pTimeBox->p_Timer;

			UInt uiNestFire = pTimer->GetNextFire();
			//cout<<"Timer Found "<<uiNestFire<<endl;
			if (uiNestFire == 0)//fire now
			{
				pTimer->SetNextFire(pTimer->GetInterval());
				BTimerCallback * pTCB = pTimer->GetTimerCB();
				pTCB->OnTimer(pTimer);//firing timer call back


				if (IsTimerLive(uiTimerId))//check for user timer deletion during call back
				{
					if (pTimer->IsRepeated()) //if repeated timer add to correct slot
					{
						pTimer->SetNextFire(pTimer->GetInterval());
						AddToCorrectSlot(pTimeBox);
					}
					else
					{
						// if timer not repeating
						pTimer->ResetTimerId();
						delete pTimeBox;
						stack_FreeTimerIDs.push(uiTimerId);
					}

				}
				else //delete during call back
				{
					delete pTimeBox;
					stack_FreeTimerIDs.push(uiTimerId);
				}

			}
			else // can not fire now
			{
				AddToCorrectSlot(pTimeBox);
			}
		}
		else //timer has been deleted by user
		{
			delete pTimeBox;
			stack_FreeTimerIDs.push(uiTimerId);
		}
		++iteAll;
	}

	lst_FireAll.clear();
	GET_CURR_TIME(tvnow,ui_CurrTime,ui_MicroSec);
	UInt uiDeleay = ui_CurrTime - ui_LastTime;
	ui_LastTime = ui_CurrTime;
	return uiDeleay;
}

BTimer * BScheduler::CreateTimer(double _dIntaval, bool _bIsRepeated, BTimerCallback * _pTCB, int _iTimerType)
{
	UInt uiTimerID = GetTimerIDFromStack();
	UInt uiInt = (UInt) (_dIntaval * 1000);

	BTimer * pTimer = new BTimer(uiInt, _pTCB, _bIsRepeated, uiTimerID, this, _iTimerType);
	ap_Timers[uiTimerID] = pTimer;
	AddToTimerList(pTimer);
	return pTimer;
}

BClient * BScheduler::CreateClient(const char * _zServerIP, UInt _uiServerPort, BClientCallback * _pCCB, bool _bHB)
{
	int hSocket = socket(AF_INET, SOCK_STREAM, 0);
	if (hSocket < 0)
	{
		//		SetError("MSOSocketMan:CreateClient: Can't create socket");
		//cout << "Socket creation failed " << endl;
		return NULL;
	}

	sockaddr_in sockAddr;
	memset(&sockAddr, 0, sizeof(sockAddr));

	sockAddr.sin_family = AF_INET;
	sockAddr.sin_addr.s_addr = inet_addr(_zServerIP);

	if (sockAddr.sin_addr.s_addr == (in_addr_t) -1)
	{
		//
		// zIPAddress may be a host name
		//
		hostent* lphost;
		lphost = gethostbyname(_zServerIP);

		if (lphost != NULL)
			sockAddr.sin_addr.s_addr = ((in_addr*) lphost->h_addr)->s_addr;
		else
		{
			//			SetError("MSOSocketMan:CreateClient: Invalid IP Address");
			//cout << "CreateClient: Invalid IP Address" << endl;
			close(hSocket);
			return NULL;
		}
	}
	//In order to remove warnings
	in_port_t inpl = 0;
	char zBuf[2];
	memcpy(zBuf, &_uiServerPort, 2);
	char zBufR[2];
	zBufR[0] = zBuf[1];
	zBufR[1] = zBuf[0];
	memcpy(&inpl, zBufR, 2);
	//	scaiAddr.sin_port = inpl;
	sockAddr.sin_port = inpl;

	// Connect to server
	if (connect(hSocket, (sockaddr*) &sockAddr, sizeof(sockAddr)) < 0)
	{
		// If a non-blocking connect() will return an of EWOULDBLOCK
		int err = errno;
		if ((err != EWOULDBLOCK) && (err != EINPROGRESS))
		{
			//cout << "MSOSocketMan:CreateClient: Couldn't connect to Server" << endl;

			close(hSocket);
			return NULL;
		}
	}
	//cout << "Connected to server" << endl;
	SetNonBlocking(hSocket);
	SetSocketOptions(hSocket);

	int iSocket = hSocket;
	struct epoll_event stEpollEvent;
	memset(&stEpollEvent, 0, sizeof(stEpollEvent));
	stEpollEvent.events = EPOLL_CLI_EVENT_FLAGS;
	stEpollEvent.data.fd = iSocket;
	if (epoll_ctl(i_EpollFd, EPOLL_CTL_ADD, iSocket, &stEpollEvent) < 0)
	{
		cout << "epoll_ctl failed.  IP " << _zServerIP << " Port " << _uiServerPort << endl;
		return NULL;
	}
	BClient * pClient = new BClient(_zServerIP, _uiServerPort, iSocket, _bHB, _pCCB, this, NULL);
	ap_EItems[iSocket] = pClient;
	if (_bHB)
		set_HBClients.insert(iSocket);
	return pClient;

}

BServer * BScheduler::CreateServer(UInt _uiPort, BServerCallback * _pSCB, bool _bHB)
{
	int iSocket = -1;
	struct sockaddr_in stAddr;

	if ((iSocket = socket(AF_INET, SOCK_STREAM, 0)) == -1)
	{
		cout << "Socket create failed. Port = " << _uiPort << endl;
		return NULL;
	}

	stAddr.sin_family = AF_INET;
	stAddr.sin_port = htons((uint16_t) _uiPort);
	stAddr.sin_addr.s_addr = htonl(INADDR_ANY);

	memset(&(stAddr.sin_zero), '\0', 8);
	SetNonBlocking(iSocket);
	SetSocketOptions(iSocket);

	if (bind(iSocket, (struct sockaddr *) &stAddr, sizeof(struct sockaddr)) == -1)
	{
		cout << "Socket bind failed. Port = " << _uiPort << endl;
		return NULL;
	}

	if (listen(iSocket, 2048) == -1)
	{
		cout << "Socket listen failed. Port = " << _uiPort << endl;
		return NULL;
	}
	struct epoll_event stEpollEvent;
	memset(&stEpollEvent, 0, sizeof(stEpollEvent));
	stEpollEvent.events = EPOLL_SVR_EVENT_FLAGS; // only care about in
	stEpollEvent.data.fd = iSocket;
	if (epoll_ctl(i_EpollFd, EPOLL_CTL_ADD, iSocket, &stEpollEvent) < 0)
	{
		cout << "epoll_ctl failed. Port = " << _uiPort << endl;
		return NULL;
	}
	BServer * pServer = new BServer(_uiPort, _bHB, iSocket, _pSCB, this);
	ap_EItems[iSocket] = pServer;
	return pServer;
}

void BScheduler::AddToTimerList(BTimer * _pTimer)
{
	UInt uiIntaval = _pTimer->GetInterval();
	int iPos, iCurrPos;
	UInt uiNext = 0;
	iCurrPos = ui_CurrentPos % MAX_ARRAY_SIZE;

	if (uiIntaval < MAX_ARRAY_SIZE) // less than max time wait intawal
	{
		uiNext = ui_CurrentPos + uiIntaval;
		iPos = uiNext % MAX_ARRAY_SIZE;
	}
	else
	{
		uiNext = ui_CurrentPos + MAX_ARRAY_SIZE - 1;
		iPos = uiNext % MAX_ARRAY_SIZE;
		_pTimer->SetNextFire(uiIntaval - MAX_ARRAY_SIZE + 1);
	}
	st_TimerBox * pTimeBox = new st_TimerBox(_pTimer->GetTimerID(), _pTimer);
	ast_Timers[iPos].push_back(pTimeBox);

}

void BScheduler::AddToCorrectSlot(st_TimerBox * _pstTimeBox)
{
	BTimer * pTimer = _pstTimeBox->p_Timer;
	UInt uiNextFire = pTimer->GetNextFire();
	int iPos, iCurrPos;
	UInt uiNext = 0;
	iCurrPos = ui_CurrentPos % MAX_ARRAY_SIZE;

	if (uiNextFire < MAX_ARRAY_SIZE) // less than max time wait intawal
	{
		uiNext = ui_CurrentPos + uiNextFire;
		iPos = uiNext % MAX_ARRAY_SIZE;
		pTimer->SetNextFire(0);
	}
	else
	{
		uiNext = ui_CurrentPos + MAX_ARRAY_SIZE - 1;
		iPos = uiNext % MAX_ARRAY_SIZE;
		pTimer->SetNextFire(uiNextFire - MAX_ARRAY_SIZE + 1);
	}
	//cout<<uiCurNext <<" Aded to "<<uiNext<<" "<<ui_CurrentPos<<endl;
	ast_Timers[iPos].push_back(_pstTimeBox);

}

void BScheduler::AddToDeletedTimerSet(UInt _uiTimerID)
{
	ap_Timers[_uiTimerID] = NULL;
	//stack_FreeTimerIDs.push(_uiTimerID);
}
void BScheduler::AddToDeletedClientSet(UInt _uiClientID)
{
	//set_DeletedClients.insert(_uiClientID);
	ap_EItems[_uiClientID] = NULL;
	if (epoll_ctl(i_EpollFd, EPOLL_CTL_DEL, _uiClientID, NULL) < 0)
	{
		cout << "Removing a Client from epoll failed.  Client = " << _uiClientID << endl;
	}
	close(_uiClientID);
	//set_HBClients.erase(_uiClientID); //remove from HB client set
}
void BScheduler::AddToDeletedServerSet(UInt _uiServerID)
{
	ap_EItems[_uiServerID] = NULL;//currently not deleting the server object. Wt to do with connected clients ????
	if (epoll_ctl(i_EpollFd, EPOLL_CTL_DEL, _uiServerID, NULL) < 0)
	{
		cout << "Removing a server from epoll failed.  Server = " << _uiServerID << endl;
	}
	close(_uiServerID);
}

bool BScheduler::IsTimerLive(UInt _uiTimerID)
{
	if (ap_Timers[_uiTimerID])
	{
		return true;
	}
	return false;
}

void BScheduler::SetNonBlocking(int _iSockID)
{
	int iOldFlags = fcntl(_iSockID, F_GETFL, 0);
	if (!(iOldFlags & O_NONBLOCK))
	{
		iOldFlags |= O_NONBLOCK;
	}
	fcntl(_iSockID, F_SETFL, iOldFlags);
}

void BScheduler::SetSocketOptions(int _iSockID)
{
	/***************************************
	 * Socket optimization options
	 ***************************************/
	unsigned int uiMask = 1;

	setsockopt(_iSockID, SOL_SOCKET, SO_REUSEADDR, &uiMask, sizeof(uiMask));

	linger ll;
	ll.l_onoff = 1;
	ll.l_linger = 1;

	setsockopt(_iSockID, SOL_SOCKET, SO_LINGER, &ll, sizeof(linger));
	uiMask = 250000;
	setsockopt(_iSockID, SOL_SOCKET, SO_SNDBUF, &uiMask, sizeof(int));
	uiMask = 250000;
	setsockopt(_iSockID, SOL_SOCKET, SO_RCVBUF, &uiMask, sizeof(int));
	uiMask = 1;
	setsockopt(_iSockID, SOL_SOCKET, SO_KEEPALIVE, &uiMask, sizeof(int));
	uiMask = 1;
	setsockopt(_iSockID, IPPROTO_TCP, TCP_NODELAY, &uiMask, sizeof(int));
}

UInt BScheduler::ReadSockets(UInt _uiInterval)
{

	if (_uiInterval > MAX_EPOL_WAIT)
		_uiInterval = MAX_EPOL_WAIT;
	int iWaitFds = epoll_wait(i_EpollFd, pstEpollRetEvents, MAX_EPOL_EVENTS, _uiInterval);

	if (iWaitFds > 0)
	{
		for (int i = 0; i < iWaitFds; ++i)
		{
			int iReadySock = pstEpollRetEvents[i].data.fd;

			BItem * pEItem = ap_EItems[iReadySock];
			if (pEItem)
			{
				if (pEItem->getEvtType() == E_EVT_SERVER) //if a servers
				{
					BServer * pTemServer = (BServer *) pEItem;
					while (true)
					{
						int iCliSock = accept(iReadySock, (struct sockaddr *) &stCliAddr, (unsigned int *) &i_CliAddrSize);
						if (iCliSock == -1)
						{
							//Client accept failed;
							break;
						}
						struct epoll_event stEpEvent;
						memset(&stEpEvent, 0, sizeof(stEpEvent));
						SetNonBlocking(iCliSock);
						SetSocketOptions(iCliSock);
						stEpEvent.events = EPOLL_CLI_EVENT_FLAGS;
						stEpEvent.data.fd = iCliSock;

						//cout << "CliSock " << iCliSock << endl;
						if (epoll_ctl(i_EpollFd, EPOLL_CTL_ADD, iCliSock, &stEpEvent) < 0)
						{
							cout << " Adding a client to epoll failed. Client = " << iCliSock << " Server = " << iReadySock << endl;
							continue;
						}

						// print client details, IP address
						sockaddr_in addr;
						memset(&addr, 0, sizeof(addr));
						unsigned int addr_len = sizeof(addr);
						const char * zCliIp = NULL;
						int iPort;

						BClient * pClient = NULL;

						if (getpeername(iCliSock, (sockaddr *) &addr, (socklen_t *) &addr_len) < 0)
						{
							pClient = new BClient(UNKNOWN_IP_ADDRESS, 0, (UInt) iCliSock, pTemServer->IsHBEnable(), NULL, this, pTemServer);
							ap_EItems[iCliSock] = pClient;
							BServerCallback * pSCB = pTemServer->GetServerCB();
							pSCB->OnConnect(pTemServer, pClient);
							pTemServer->AddtoConnectedClientsList(pClient);
						}
						else
						{
							zCliIp = inet_ntoa(addr.sin_addr);
							iPort = addr.sin_port;
							pClient = new BClient(zCliIp, (UInt)iPort, (UInt) iCliSock, pTemServer->IsHBEnable(), NULL, this, pTemServer);
							ap_EItems[iCliSock] = pClient;
							BServerCallback * pSCB = pTemServer->GetServerCB();
							pSCB->OnConnect(pTemServer, pClient);
							pTemServer->AddtoConnectedClientsList(pClient);
						}
					}
				}
				else // not a server then client
				{
					int iCliId = iReadySock;
					BClient * pClient = (BClient *) pEItem;
					if (pstEpollRetEvents[i].events & EPOLLRDHUP)
					{

						BServer * pServer = pClient->GetServer();
						if (!pServer)
						{
							BClientCallback * pCCB = pClient->GetClientCB();
							ap_EItems[iCliId] = NULL;
							pCCB->OnDisconnect(pClient, EVTERR_NONE);
						}
						else
						{
							BServerCallback * pSCB = pServer->GetServerCB();
							ap_EItems[iCliId] = NULL;
							pSCB->OnDisconnect(pServer, pClient, EVTERR_NONE);
							pServer->RemoveFromConnectedClientsList(pClient);
							//pServer->RemoveFromClientSet(iCliId);
						}

						continue;
					}
					if (pstEpollRetEvents[i].events & EPOLLIN)
					{
						int iReadByteCount = 0;

						//do // is this good to read until no data avail. What if malform-client????
						for (int t = 0; t < i_ReadTurns; ++t)
						{
							pClient = (BClient *) ap_EItems[iCliId];
							if (pClient)
							{
								// read error handling
								if ((iReadByteCount = read(iCliId, z_ReadBuff, CLI_READ_SIZE)) > 0)
								{
									BServer * pServer = pClient->GetServer();
									if (!pServer)
									{
										//cout << "Ondata" << z_ReadBuff << endl;
										BClientCallback * pCCB = pClient->GetClientCB();
										WriteSockLog(true, iCliId, z_ReadBuff, iReadByteCount);
										pCCB->OnData(pClient, z_ReadBuff, iReadByteCount);
									}
									else
									{
										BServerCallback * pSCB = pServer->GetServerCB();
										WriteSockLog(true, iCliId, z_ReadBuff, iReadByteCount);
										pSCB->OnData(pServer, pClient, z_ReadBuff, iReadByteCount);
									}

								}
								else if (iReadByteCount == 0)
								{
									BServer * pServer = pClient->GetServer();
									if (!pServer)
									{
										BClientCallback * pCCB = pClient->GetClientCB();
										ap_EItems[iCliId] = NULL;
										pCCB->OnDisconnect(pClient, EVTERR_NONE);
									}
									else
									{
										BServerCallback * pSCB = pServer->GetServerCB();
										ap_EItems[iCliId] = NULL;
										pSCB->OnDisconnect(pServer, pClient, EVTERR_NONE);
										//pServer->RemoveFromClientSet(iCliId);
									}
									break;
								}
								else
								{
									if (errno != EAGAIN)
									{
										BServer * pServer = pClient->GetServer();
										if (!pServer)
										{
											BClientCallback * pCCB = pClient->GetClientCB();
											ap_EItems[iCliId] = NULL;
											pCCB->OnDisconnect(pClient, EVTERR_NONE);
										}
										else
										{
											BServerCallback * pSCB = pServer->GetServerCB();
											ap_EItems[iCliId] = NULL;
											pSCB->OnDisconnect(pServer, pClient, EVTERR_NONE);
											//pServer->RemoveFromClientSet(iCliId);
										}
									}
									break;
								}
							}
							else
							{
								break;
							}
						}

					}
					if (pstEpollRetEvents[i].events & EPOLLOUT) //can write to this client
					{
						set_OnReadyToSendClients.insert(iCliId);
						pClient->SetBufferStatus(false);
						int iBufLen = pClient->m_SendBuffer.GetBufLen();
						int iSendLen = SendToClient(iCliId);
						if (iBufLen == iSendLen)
						{
							struct epoll_event stEpEvent;
							memset(&stEpEvent, 0, sizeof(stEpEvent));
							stEpEvent.events = EPOLL_CLI_EVENT_FLAGS;
							stEpEvent.data.fd = iCliId;

							if (epoll_ctl(i_EpollFd, EPOLL_CTL_MOD, iCliId, &stEpEvent) < 0)
							{
								cout << "Modifying epoll failed. Client = " << iCliId << endl;
							}

						}
					}

				}
			}

		}
	}
	if (set_OnReadyToSendClients.size() > 0)
	{
		ONR2S_CLIENTS::iterator ite = set_OnReadyToSendClients.begin();
		while (ite != set_OnReadyToSendClients.end())
		{
			int iClientId = *ite;
			BClient * pClient = (BClient *) ap_EItems[iClientId];
			if (pClient)
			{

				BServer * pServer = pClient->GetServer();
				if (!pServer)
				{
					BClientCallback * pCCB = pClient->GetClientCB();
					pCCB->OnReadyToSend(pClient);
				}
				else
				{
					BServerCallback * pSCB = pServer->GetServerCB();
					pSCB->OnReadyToSend(pServer, pClient);
				}
			}
			++ite;
		}
		set_OnReadyToSendClients.clear();
	}
	if (set_DiconClients.size() > 0)
	{
		WRITE_DESCON_CLIENTS::iterator ite = set_DiconClients.begin();
		while (ite != set_DiconClients.end())
		{
			int iClientId = *ite;
			BClient * pClient = (BClient *) ap_EItems[iClientId];

			if (pClient)
			{
				BServer * pServer = pClient->GetServer();
				if (!pServer)
				{
					BClientCallback * pCCB = pClient->GetClientCB();
					ap_EItems[iClientId] = NULL;
					pCCB->OnDisconnect(pClient, EVTERR_NONE);
				}
				else
				{
					BServerCallback * pSCB = pServer->GetServerCB();
					ap_EItems[iClientId] = NULL;
					pSCB->OnDisconnect(pServer, pClient, EVTERR_NONE);
				}
			}
			else
			{
				if (epoll_ctl(i_EpollFd, EPOLL_CTL_DEL, iClientId, NULL) < 0)
				{
					cout << "Removing a client from epoll failed.  Client = " << iClientId << endl;
				}
			}
			++ite;
		}
		set_DiconClients.clear();
	}
	GET_CURR_TIME(tvnow,ui_CurrTime,ui_MicroSec);
	UInt uiDeleay = ui_CurrTime - ui_LastTime;
	ui_LastTime = ui_CurrTime;
	return uiDeleay;
}

int BScheduler::SendToClient(UInt _uiClid)
{
	int iClientId = _uiClid;

	BClient * pClient = (BClient *) ap_EItems[iClientId];
	if (pClient)
	{
		int iCurSend = 0, iSendLen = 0;
		int iBufLen = pClient->m_SendBuffer.GetBufLen();
		const char * zBuf = pClient->m_SendBuffer.GetBuffer();

		int iTotLen = iBufLen;
		while (iSendLen < iTotLen)
		{
			if ((iCurSend = write(iClientId, zBuf + iSendLen, iBufLen)) < 0)
			{
				if (errno == EAGAIN && iBufLen > 0)
				{
					// buffered at socket level.
					pClient->SetBufferStatus(true);
					WriteSockLog(false, _uiClid, zBuf, iSendLen);
					pClient->m_SendBuffer.DeleteFromStart(iSendLen);

					// add OUT to epoll
					struct epoll_event stEpEvent;
					stEpEvent.events = EPOLL_CLI_EVENT_FLAGS | EPOLLOUT;
					stEpEvent.data.fd = iClientId;

					if (epoll_ctl(i_EpollFd, EPOLL_CTL_MOD, iClientId, &stEpEvent) < 0)
					{
						cout << "Modifying epoll failed. Client = " << iClientId << endl;
					}
					return iSendLen;
				}
				else
				{
					// write error - disconnect
					WriteSockLog(false, _uiClid, zBuf, iSendLen);
					pClient->m_SendBuffer.DeleteFromStart(iSendLen);
					set_DiconClients.insert(iClientId);
					return 0;
				}
			}

			iBufLen -= iCurSend;
			iSendLen += iCurSend;
		}

		WriteSockLog(false, _uiClid, zBuf, iSendLen);
		//pClient->m_SendBuffer.DeleteFromStart(iSendLen);
		pClient->m_SendBuffer.Reset();
		return iSendLen;

	}
	return -1;

}
int BScheduler::GetTimerIDFromStack()
{
	if (!stack_FreeTimerIDs.empty()) //if not empty
	{
		int iID = stack_FreeTimerIDs.top();
		stack_FreeTimerIDs.pop();
		return iID;
	}
	else
	{
		BTimer ** apAldTimers = ap_Timers;
		ap_Timers = new BTimer*[i_MaxTimers + MAX_TIMERS_INCREASE_INTERVAL];
		for (int i = 0; i < i_MaxTimers; ++i) //copy old timers to new array
		{
			ap_Timers[i] = apAldTimers[i];
		}
		int iPreMax = i_MaxTimers;
		i_MaxTimers += MAX_TIMERS_INCREASE_INTERVAL;
		for (int i = iPreMax; i < i_MaxTimers; ++i) //copy old timers to new array
		{
			ap_Timers[i] = NULL;
			stack_FreeTimerIDs.push(i);
		}
		delete[] apAldTimers;
		int iID = stack_FreeTimerIDs.top();
		stack_FreeTimerIDs.pop();
		return iID;
	}
}
//void BScheduler::OnTimer(BTimer * _pTimer)
//{
//	ALL_HB_CLIENTS::iterator ite = set_HBClients.begin();
//	while (ite != set_HBClients.end())
//	{
//		BClient * pClient = (BClient *) ap_EItems[*ite];
//		if (pClient)
//		{
//			pClient->Send(z_HeartBeat, i_HBLen);
//			++ite;
//		}
//		else
//		{
//			ALL_HB_CLIENTS::iterator iteTem = ite;
//			++ite;
//			set_HBClients.erase(iteTem);
//			continue;
//			_pTimer = _pTimer;
//		}
//	}
//}
void BScheduler::CreateLog(const char * _zLogFileName)
{
	if (i_SockLogFd < 0) // file NOT exist
	{
		strcpy(z_SockLogPath, _zLogFileName);
		i_SockLogFd = open(z_SockLogPath, O_TRUNC | O_CREAT | O_WRONLY, S_IREAD | S_IWRITE | S_IRGRP | S_IROTH);
	}
	b_ISSocketLogON = true;
}
void BScheduler::SetLogON()
{
	if (z_SockLogPath[0] != 0)
	{
		CreateLog(z_SockLogPath);
	}
	else
		b_ISSocketLogON = false;
}
void BScheduler::SetLogOff()
{
	b_ISSocketLogON = false;
}
void BScheduler::WriteSockLog(bool _bIn, int _iCliId, const char* _zBuff, int _iBufLen)
{
	if (b_ISSocketLogON)
	{
		time_t tp;
		struct tm stTm;
		timeval tv;

		gettimeofday(&tv, NULL);

		if (tv.tv_sec != i_LocalSeconds)
		{
			time(&tp);
			strftime(z_LocalTimeStamp, 25, "%Y%m%d-%H:%M:%S.", localtime_r(&tp, &stTm));
			i_LocalSeconds = tv.tv_sec;
		}

		sprintf(z_LocalTimeStamp + 18, "%06ld", tv.tv_usec);
		z_LocalTimeStamp[21] = ':';
		z_LocalTimeStamp[22] = 0;

		const char * zFmt = _bIn ? "%s INPUT  %05d : " : "%s OUTPUT %05d : ";
		sprintf(z_SockLogBuff, zFmt, z_LocalTimeStamp, _iCliId);
		if (!b_ISSocketLogDump)
			write(i_SockLogFd, z_SockLogBuff, 38);
		write(i_SockLogFd, _zBuff, _iBufLen);
	}
}
void BScheduler::Detach(BClient * _pClient)
{
	int iClid = _pClient->GetClientID();
	if (epoll_ctl(i_EpollFd, EPOLL_CTL_DEL, iClid, NULL) < 0)
	{
		cout << "[Detach] Removing a Client from epoll failed.  Client = " << iClid << " Error:[" << errno << "] " << strerror(errno) << endl;
	}
	ap_EItems[iClid] = NULL;
	_pClient->SetEventSheduler(NULL);
	_pClient->SetCCB(NULL);
	_pClient->SetServer(NULL);
	if (_pClient->IsHBEnable())
		set_HBClients.erase(iClid);
}
void BScheduler::AttachClient(BClient * _pClient, BClientCallback * _pCCB)
{
	int iClid = _pClient->GetClientID();
	ap_EItems[iClid] = _pClient;
	_pClient->SetCCB(_pCCB);
	_pClient->SetEventSheduler(this);
	struct epoll_event stEpollEvent;
	memset(&stEpollEvent, 0, sizeof(stEpollEvent));
	if (_pClient->IsBuffered())
		stEpollEvent.events = EPOLL_CLI_EVENT_FLAGS | EPOLLOUT;
	else
		stEpollEvent.events = EPOLL_CLI_EVENT_FLAGS;
	stEpollEvent.data.fd = iClid;
	if (epoll_ctl(i_EpollFd, EPOLL_CTL_ADD, iClid, &stEpollEvent) < 0)
	{
		cout << "[AttachClient] epoll_ctl failed. CLID " << iClid << " Error:[" << errno << "] " << strerror(errno) << endl;
	}
	if (_pClient->IsHBEnable())
		set_HBClients.insert(iClid);
}

void BMulticastSender::CheckAndSetMulticastInfo(UInt _uiMCGroupIP, UShort _usMCPort)
{
	if (ui_MulticastGroupIP != _uiMCGroupIP)
	{
		IPAddrConverter oIPConv;
		char zTempIp[100];
		oIPConv.GetIPAddr(zTempIp, _uiMCGroupIP);
		strcpy(z_MulticastGroupIP, zTempIp);
		ui_MulticastGroupIP = _uiMCGroupIP;
		if (i_IntendedPort != (int) _usMCPort)
		{
			i_IntendedPort = _usMCPort;
		}
	}

}

